package cn.itcast.function;


import cn.itcast.bean.StockBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.util.HbaseUtil;
import cn.itcast.util.QuotUtil;
import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.StringUtils;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.List;

/**
 * 封装hbase批量写入对象List<Put>
 */
public class SecStockHbaseFunction implements AllWindowFunction<StockBean, List<Put>, TimeWindow> {
    @Override
    public void apply(TimeWindow timeWindow, Iterable<StockBean> iterable, Collector<List<Put>> collector) throws Exception {
        /**
         * 1.新建List<Put>
         * 2.循环数据
         * 3.设置rowkey
         * 4.json数据转换
         * 5.封装put
         * 6.收集数据
         */

        List<Put> list = new ArrayList<>();
        String rowkey;
        for (StockBean stockBean : iterable) {
            rowkey=stockBean.getSecCode()+stockBean.getTradeTime();
            //将对象转换为json字符串，放入hbase中
            String str = JSON.toJSONString(stockBean);
            //封装put
            Put put = new Put(Bytes.toBytes(rowkey));
            put.addColumn("info".getBytes(),"data".getBytes(),str.getBytes());

            list.add(put);
        }
        collector.collect(list);
    }
}
